#include "Handlers.h"
#include "Settings.h"
#include "CMSInterface.h"



std::vector<AgentWorker*>  AgentWorker::g_VecAgent;
ACE_Thread_Mutex	AgentWorker::g_AgentLock;

std::vector<_JOB_INFO> AgentWorker::g_JobList;
ACE_Thread_Mutex AgentWorker::g_JobLock;	


int AgentWorker::handle_close (ACE_HANDLE h, ACE_Reactor_Mask mask)
{
	if(_STOPPED != state && !timeout)
	{
		LOG::PushProcessLog(_ERROR , GetRemoteAddr() + " : agent connection closed abnormally");
	}

	delete this;
	return 0;
}


bool AgentWorker::IsAgentSlotFull()
{
	bool full = false;
	g_AgentLock.acquire();

	if(g_VecAgent.size() + 1 > Settings::GetSettings()->MAX_AGENT)
	{
		full = true;

		char buf[1024];
		sprintf(buf , "agent slot is full : %d/%d" , g_VecAgent.size() , Settings::GetSettings()->MAX_AGENT);
		LOG::PushProcessLog(_ERROR, buf);

	}

	g_AgentLock.release();

	return full;
}

bool AgentWorker::IsDuplicatedAddr(const std::string& addr)
{
	bool find = false;

	g_AgentLock.acquire();

	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{

		if(g_VecAgent[i]->GetRemoteAddr() == addr)
		{
			find = true;

			char buf[1024];
			sprintf(buf , "duplicated connection from  : %s" , addr.c_str() );
			LOG::PushProcessLog(_ERROR, buf);
			break;
		}

	}
	g_AgentLock.release();

	return find;
}



bool AgentWorker::AddAgent(AgentWorker * agent)
{
	bool ret = true;

	ACE_ASSERT(agent);
	char buf[1024];

	_DB_RESULT db_result = DBInterface::AddAgent(Settings::GetSettings()->RENDERER_ID , agent->GetRemoteAddr());
	if(_DB_SUCCESS != db_result) 
	{
		ret = false;
		sprintf(buf , "DB::AddAgent  :%d ,  %s" ,db_result ,  agent->GetRemoteAddr().c_str());
		printf("%s\n" , buf);
		LOG::PushProcessLog(_ERROR, buf);
	}
	else
	{
		g_AgentLock.acquire();
		g_VecAgent.push_back(agent);
		g_AgentLock.release();

		sprintf(buf , "add agent : %d/%d => %d/%d" , g_VecAgent.size()-1 , Settings::GetSettings()->MAX_AGENT ,g_VecAgent.size() , Settings::GetSettings()->MAX_AGENT);
		printf("%s\n" , buf);
		LOG::PushProcessLog(_INFO, buf);
	}


	return ret;
}

bool AgentWorker::DelAgent(AgentWorker * agent)
{
	bool ret = true;
	ACE_ASSERT(agent);
	char buf[1024];

	g_AgentLock.acquire();

	ACE_ASSERT(find(g_VecAgent.begin() , g_VecAgent.end() , agent) != g_VecAgent.end());

	std::vector<AgentWorker*>::iterator pos = find(g_VecAgent.begin() , g_VecAgent.end() , agent);
	if(pos != g_VecAgent.end())
	{
		g_VecAgent.erase(pos);
	}
	else
	{
		ret = false;
	}

	g_AgentLock.release();

	sprintf(buf , "del agent : %d/%d => %d/%d" , g_VecAgent.size()+1, Settings::GetSettings()->MAX_AGENT,g_VecAgent.size(), Settings::GetSettings()->MAX_AGENT);
	printf("%s\n" , buf);

	_DB_RESULT db_result = DBInterface::DelAgent(Settings::GetSettings()->RENDERER_ID ,agent->GetRemoteAddr()); 
	if(_DB_SUCCESS != db_result) 
	{
		sprintf(buf , "DB::DelAgent  :%d ,  %s" ,db_result ,  agent->GetRemoteAddr().c_str());
		printf("%s\n" , buf);
		LOG::PushProcessLog(_ERROR, buf);
	}
	else
	{
		LOG::PushProcessLog(_INFO , buf);
	}


	if(agent->m_vecReservedJob.size() > 0)
	{
		PushCanceledJobs(agent->m_vecReservedJob);
	}

	
	return ret;
}

bool AgentWorker::IsWorking(const std::string& job_id)
{
	bool ret = false;

	m_workerLock.acquire();
	for(int j = 0 ; j < m_vecWorks.size() ; j ++)
	{
		if(m_vecWorks[j].GetJobId() ==  job_id)
		{
			ret = true;
			break;
		}
	}
	m_workerLock.release();

	return ret;

}

bool AgentWorker::AddWork()
{
	bool ret = false;

	_JOB_INFO ji;

	m_workerLock.acquire();
	if(GetMaxWorkCount() > m_vecWorks.size())
	{
		while(true)
		{
			ji = FetchJob();
			if(ji.IsValid())
			{
				bool same = false;
				for(int i = 0 ; i < m_vecWorks.size(); i ++)
				{
					if(ji.GetJobId() == m_vecWorks[i].GetJobId() )
					{
						same = true;
						break;
					}
				}
				if(!same)
				{
					m_vecWorks.push_back(ji);
					m_vecReservedJob.push_back(ji);
					ret = true;
					break;
				}
			}
			else
			{
				break;
			}

		} 

	} 

	m_workerLock.release();

	if(ret)
	{
		LOG::PushJobLog(_REQUIRED , ji.GetJobId() , GetRemoteAddr());

		PACKET_WORK w(ji);
		Send(w);

	}

	return ret;

}





int AgentWorker::GetSpareCount()
{
	if(GetMaxWorkCount() - m_vecWorks.size() >= 0)
		return GetMaxWorkCount() - m_vecWorks.size();

	return 0;
}

int AgentWorker::GetWorkCount()
{
	return m_vecWorks.size();
}

int AgentWorker::GetReservedCount()
{
	return m_vecReservedJob.size();
}

int AgentWorker::GetTotalWorkCount() 
{
	int tot_count = 0;
	g_AgentLock.acquire();

	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{
		AgentWorker * a = g_VecAgent[i];
		if(a)
		{
			tot_count  += a->GetWorkCount();
		}
	}

	g_AgentLock.release();

	return tot_count;

}

AgentWorker * AgentWorker::GetRichestWorker()
{
	int spare_count = 0;
	AgentWorker * richest_worker = NULL;

	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{
		AgentWorker * a = g_VecAgent[i];
		if(a && a->state == _RUNNING) 
		{
			if(spare_count  < a->GetSpareCount())
			{
				spare_count = a->GetSpareCount();
				richest_worker = a;
			}
		}
	}

	return richest_worker;
}


int AgentWorker::GetTotalSpareCount()
{
	int spare_count = 0;

	g_AgentLock.acquire();
	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{
		if(g_VecAgent[i] && g_VecAgent[i]->state == _RUNNING)
		{
			spare_count += g_VecAgent[i]->GetSpareCount();
		}
	}
	g_AgentLock.release();

	return spare_count;
}

int AgentWorker::GetAgentCount()
{
	int count = 0;

	g_AgentLock.acquire();

	count = g_VecAgent.size();

	g_AgentLock.release();

	return count;

}


void AgentWorker::GetJobList()
{
	int size = 0;

	g_JobLock.acquire();
	size = g_JobList.size();
	g_JobLock.release();

	if(size < Settings::GetSettings()->MIN_JOB_LIST)
	{
		std::vector<_JOB_INFO> new_list;

		if(CMSInterface::GetJobList(Settings::GetSettings()->MAX_JOB_LIST - size , new_list) > 0)
		{
			if(new_list.size() > 0)
			{
				g_JobLock.acquire();

				printf("Get Job List %d + %d = %d\n" , g_JobList.size() , new_list.size() , g_JobList.size() + new_list.size());
				g_JobList.insert(g_JobList.end() , new_list.begin() , new_list.end());

				g_JobLock.release();
			}


		}

	}

}

void AgentWorker::PushCanceledJobs(const std::vector<_JOB_INFO>& vecCancel)
{
	g_JobLock.acquire();

	g_JobList.insert(g_JobList.begin() , vecCancel.begin() , vecCancel.end());

	g_JobLock.release();
}


_JOB_INFO AgentWorker::FetchJob()
{
	_JOB_INFO ji;

	g_JobLock.acquire();

	if(g_JobList.size() > 0)
	{
		ji = g_JobList[0];
		g_JobList.erase(g_JobList.begin());

		printf("job list size : %d\n" , g_JobList.size());
	}

	g_JobLock.release();

	return ji;

}



bool AgentWorker::Execute(const std::string &rcv_text)
{


	Json::Value root;
	if(!PACKET_BASE::GetRoot(rcv_text , root))
	{
		root.clear();
		return false;
	}

	std::string sCmd = PACKET_BASE::GetStringS(root , "CMD");
	std::string renderer_id = PACKET_BASE::GetStringS(root , "RENDERER_ID");

	if(Settings::GetSettings()->RENDERER_ID != renderer_id)
	{
		printf("[controller] Different ID this : %s , agent : %s => close connection\n" , Settings::GetSettings()->RENDERER_ID.c_str() , renderer_id.c_str());

		root.clear();
		m_Stream.close();

		return false;
	}


	if(strcmp(sCmd.c_str() , _KEY_TICK)==0)
	{
		DistJobs();
	}
	else if(strcmp(sCmd.c_str() , _KEY_AGENT)==0)
	{
		PACKET_AGENT p(root);

		m_nAgentCpu = p.GetCPU();

		char buf[1024];
		sprintf(buf , "%s CPU count : %d , Max job count : %d " ,GetRemoteAddr().c_str() , m_nAgentCpu ,GetMaxWorkCount());
		printf("%s\n" , buf);
		LOG::PushProcessLog(_INFO, buf);
		
	}
	else if(strcmp(sCmd.c_str() , _KEY_WORK_END)==0)
	{
		PACKET_WORK_END we(root);
		return FinishWork(we.GetJobId());

	}
	else if(strcmp(sCmd.c_str() , _KEY_END_LIST)==0)
	{
		PACKET_END_LIST el(root);
		return DelReservedJobs(root);
	}


	return true;
}

bool AgentWorker::DelReservedJobs(const Json::Value& root)
{

	bool find = false;

	m_workerLock.acquire();

	printf("RESERVED %d (DEL : %d) " , m_vecReservedJob.size() , root["jobIdList"].size());

	for(int ri = 0 ; ri < root["jobIdList"].size() ; ri ++)
	{
		std::string job_id = root["jobIdList"][ri].asCString();
		find = false;

		for(int j = 0 ; j < m_vecReservedJob.size() ; j ++)
		{
			if(m_vecReservedJob[j].GetJobId() == job_id)
			{
				find = true;
				m_vecReservedJob.erase(m_vecReservedJob.begin() + j);
				break;
			}
		}

		if(!find)
		{
			printf("FAIL TO DEL RESERVED : %s\n" , job_id.c_str());
			break;
		}
	}

	printf("=> %d\n" , m_vecReservedJob.size());

	m_workerLock.release();

	return find;
}



bool AgentWorker::FinishWork(const std::string& job_id)
{

	bool ret = false;

	m_workerLock.acquire();

	for(int j = 0 ; j < m_vecWorks.size() ; j ++)
	{
		if(m_vecWorks[j].GetJobId() == job_id)
		{
			m_vecWorks.erase(m_vecWorks.begin() + j);

			ret = true;
			break;
		}
	}

	m_workerLock.release();


	DistJobs();


	ACE_ASSERT(ret);

	return true;
}



int AgentWorker::SendReadSetting()
{
	PACKET_READ_SETTINGS rs;
	return Send(rs);
}

bool AgentWorker::SendAllReadSetting()
{
	g_AgentLock.acquire();

	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{
		AgentWorker * a = g_VecAgent[i];
		if(a)
		{
			a->SendReadSetting();
		}
	}

	g_AgentLock.release();

	return true;

}



void AgentWorker::CheckAgentState()
{
	std::map<std::string , _SOCKET_STATE> agent_map;
	std::map<std::string,_SOCKET_STATE>::iterator it;

	_DB_RESULT db_result = DBInterface::GetAgentStates(Settings::GetSettings()->RENDERER_ID , agent_map);
	if(_DB_SUCCESS != db_result) 
	{
		char buf[1024];
		sprintf(buf , "DB::GetAgentStates  :%d " ,db_result);
		printf("%s\n" , buf);
		LOG::PushProcessLog(_ERROR, buf);

		return;
	}

	g_AgentLock.acquire();

	for(int i = 0 ; i < g_VecAgent.size() ; i ++)
	{
		AgentWorker * a = g_VecAgent[i];
		if(a)
		{
			it = agent_map.find(a->GetRemoteAddr());
			if( it != agent_map.end())
			{
				if(a->state != it->second)
				{
					printf("%s status changed %d ->%d\n" , a->GetRemoteAddr().c_str() , a->state , it->second);
					a->state = it->second;
				}

				if(a->state == _STOPPING && a->GetReservedCount() == 0)
				{
					printf("agent [%s] stopped \n" , a->GetRemoteAddr().c_str());
					a->state = _STOPPED;
					_DB_RESULT db_result = DBInterface::SetStoppedAgent(Settings::GetSettings()->RENDERER_ID , a->GetRemoteAddr());
					if(_DB_SUCCESS != db_result) 
					{
						char buf[1024];
						sprintf(buf , "DB::SetStoppedAgent  :%d ,  %s" ,db_result ,  a->GetRemoteAddr().c_str());
						printf("%s\n" , buf);
						LOG::PushProcessLog(_ERROR, buf);
					}
				}
			}
			else if(a->state != _UNREGISTERED) 
			{
				a->state = _UNREGISTERED;
				printf("close unregistered agent : %s \n" , a->GetRemoteAddr().c_str());
				LOG::PushProcessLog(_ERROR , "unregistered agent : " + a->GetRemoteAddr());
				a->m_Stream.close();
			}
		}

	}

	g_AgentLock.release();

}

bool AgentWorker::DistJobs()
{
	if(Settings::GetSettings()->OPERATION_MODE == 'O' || Settings::GetSettings()->OPERATION_MODE == 'o')
	{
		if(state == _RUNNING)
		{
			while(AddWork());
		}
	}
	
	return true;
}

void SettingCheckThread::CheckControllerTime()
{
	int p_t , s_t;

	_DB_RESULT db_result= DBInterface::GetControllerTimeElapsed(Settings::GetSettings()->RENDERER_ID , p_t , s_t);
	if(_DB_SUCCESS != db_result)
	{
		printf("monitoring db error(get time) : exit\n");
		LOG::SendSMS("monitoring db error(get time) : exit",_ERROR);
		exit(0);
	}

	if(Settings::GetSettings()->IsPrimary())
	{
		if(p_t > s_t)
		{
			printf("secondary is more recent : exit\n");
			LOG::SendSMS("secondary is more recent : exit",_ERROR);
			exit(0);
		}
		else if(_DB_SUCCESS !=	DBInterface::SetControllerTime(Settings::GetSettings()->RENDERER_ID , true))
		{
			printf("monitoring db error(set primary time) : exit\n");
			LOG::SendSMS("monitoring db error(set primary time) : exit",_ERROR);
			exit(0);
		}
	}
	else
	{
		if(p_t <= s_t)
		{
			printf("primary is more recent : exit\n");
			LOG::SendSMS("primary is more recent : exit",_ERROR);
			exit(0);
		}
		else if(_DB_SUCCESS != DBInterface::SetControllerTime(Settings::GetSettings()->RENDERER_ID , false))
		{
			printf("monitoring db error(set secondary time) : exit\n");
			LOG::SendSMS("monitoring db error(set secondary time) : exit",_ERROR);
			exit(0);
		}
	}
}

int SettingCheckThread::svc()
{
	char prev_operation_mode;
	int check_count = 0;

	time_t sms_tick_time = time(NULL);

	while(!stop)
	{
		if(time(NULL) - sms_tick_time > Settings::GetSettings()->PUSH_ALIVE_TICK_DELAY * 60)
		{
			char buf[100];
			sprintf(buf , "ALIVE!! agent count : %d" , AgentWorker::GetAgentCount());

			LOG::SendSMS(buf);
			sms_tick_time = time(NULL);
		}

		if(check_count % 2 == 0)
		{
			CheckControllerTime();
		}

		if(check_count ++ == Settings::GetSettings()->SETTING_CHECK_COUNT)
		{
			check_count = 0;

			prev_operation_mode = Settings::GetSettings()->OPERATION_MODE;
			if(Settings::ApplyChangedSettingValue())
			{
				LOG::SendSMS("server setting has been changed");

				printf("\n#################### Changed Setting Values ################\n");
				printf("%s" , Settings::GetSettings()->dump().c_str());
				printf("###############################################################\n");

				if(!Settings::GetSettings()->IsControllerAddr())
				{
					printf("[controller] Invalid controller for this zone\n");
					LOG::SendSMS("Invalid controller for "+Settings::GetSettings()->RENDERER_ID,_ERROR);
					ACE_Reactor::instance ()->end_event_loop();
					exit(0); 
				}
			

					AgentWorker::SendAllReadSetting();

			}

			switch(Settings::GetSettings()->OPERATION_MODE)
			{
			case 'S':
			case 's':
				if(	prev_operation_mode != Settings::GetSettings()->OPERATION_MODE) 
				{
					printf("[controller] operation mode : stop \n");
				}
				break;
			case 'K':
			case 'k':
				if(	prev_operation_mode != Settings::GetSettings()->OPERATION_MODE) 
				{
					printf("[controller] operation mode : kill process\n");
					LOG::SendSMS("kill option : "  +Settings::GetSettings()->RENDERER_ID);
				}
				ACE_Reactor::instance ()->end_event_loop();
				exit(0); 
				break;
			default: 
				if(	prev_operation_mode != Settings::GetSettings()->OPERATION_MODE) 
				{
					printf("[controller] operation mode : restart \n");
				}

			}
		}

		AgentWorker::CheckAgentState();
		
		sleep(Settings::GetSettings()->JOB_CHECK_DELAY);

	} 

	return 0;

}


int GetJobThread::svc()
{
	while(!stop)
	{
		AgentWorker::GetJobList();
		sleep(Settings::GetSettings()->JOB_CHECK_DELAY);
	}

	return 0;

}


AcceptHandler:: AcceptHandler(ACE_Reactor *reactor) :
        ACE_Event_Handler(),
        mReactor(reactor == 0 ? ACE_Reactor::instance() : reactor),
        mAcceptor() {
    
}

AcceptHandler::~AcceptHandler() {
    
}

int AcceptHandler::open(void) {
    

	ACE_INET_Addr addr(Settings::GetSettings()->CONTROLLER_PORT);

    if (mAcceptor.open(addr, 1) == -1)
      ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("%N:%l: Failed to open ")
                        ACE_TEXT ("listening socket. (errno = %i: %m)\n"), ACE_ERRNO_GET), -1);

    if (mReactor->register_handler(this,
            ACE_Event_Handler::ACCEPT_MASK) == -1) {
      ACE_ERROR((LM_ERROR, ACE_TEXT("%N:%l: Failed to register accept ")
                 ACE_TEXT ("handler. (errno = %i: %m)\n"), ACE_ERRNO_GET));

        if (mAcceptor.close() == -1)
          ACE_ERROR((LM_ERROR, ACE_TEXT("%N:%l: Failed to close the socket ")
                     ACE_TEXT ("after previous error. (errno = %i: %m)\n"),
                        ACE_ERRNO_GET));
        return -1;
    }

    return 0;
}

ACE_HANDLE AcceptHandler::get_handle(void) const 
{
    ACE_TRACE("AcceptHandler::get_handle(void)");
    return mAcceptor.get_handle();
}

int AcceptHandler::handle_input(ACE_HANDLE) 
{
	ACE_INET_Addr inet_addr;
	ACE_SOCK_Stream stream;

    if (mAcceptor.accept(stream , &inet_addr) == -1)
	{
		return 0;
	}

	if(AgentWorker::IsAgentSlotFull())
	{
		stream.close();
		return 0;
	}

	std::string addr;
	if( 0 == stream.get_remote_addr(inet_addr))
	{
		char buf[256];
		inet_addr.addr_to_string(buf , 256);

		std::string ret = buf;
		int find_pos = ret.find(":");
		if(find_pos  > 0)
		{
			addr = ret.substr(0 , find_pos);
		}
	}

	if(AgentWorker::IsDuplicatedAddr(addr))
	{
		stream.close();
		return 0;
	}

	AgentWorker * pWorker = new AgentWorker();
	pWorker->SetStream(mReactor , addr , stream); 

	if(!AgentWorker::AddAgent(pWorker))
	{
		stream.close();
		delete pWorker;
		return 0;
	}

    if (mReactor->register_handler(pWorker , ACE_Event_Handler::READ_MASK) == -1)
	{
		char buf[1024];
		sprintf(buf , "can not register reactor : %s -> close\n" , addr.c_str());
		LOG::PushProcessLog(_ERROR, buf);

		stream.close();
		delete pWorker;
		return 0;
	}


    return 0; 
}

int AcceptHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {


    if (mAcceptor.close() == -1)
      ACE_ERROR((LM_ERROR, ACE_TEXT("%N:%l: Failed to close the ")
                 ACE_TEXT ("socket. (errno = %i: %m)\n"), ACE_ERRNO_GET));

    delete this;
    return 0;
}
